
#include "include/task_dispatcher_impl.h"

#include <functional>
#include <iostream>

#include "include/thread_pool.h"

namespace dispatcher
{

class DispatcherStateManager
{
public:
    enum DispatcherState : uint8_t
    {
        NotStart = 0x00,
        Starting = 0x01,
        Started  = 0x02,
        Quiting  = 0x03,
        Quitted  = 0x04,
    };

public:
    explicit DispatcherStateManager() : state_(NotStart)
    {
    }

    void SetState(DispatcherState state)
    {
        if (state_ != state)
        {
            echo(state);
            state_ = state;
        }
    }

    DispatcherState State() const
    {
        return state_.load();
    }

protected:
    void echo(DispatcherState state)
    {
        auto map_to_string = [](DispatcherState state) -> std::string const {
            switch (state)
            {
                case NotStart:
                    return "NotStart";
                case Starting:
                    return "Starting";
                case Started:
                    return "Started";
                case Quiting:
                    return "Quiting";
                case Quitted:
                    return "Quitted";
                default:
                    break;
            }
        };
        std::cout << "[ \e[1;32mINFO\e[0m ]: Dispatcher State changed. src=" << map_to_string(state_)
                  << ", dst=" << map_to_string(state) << std::endl;
    }

private:
    std::atomic<DispatcherState> state_;
};

TaskDispatcherImpl::TaskDispatcherImpl()
  : state_manager_sptr_(std::make_shared<DispatcherStateManager>())
  , dispatcher_thread_(std::bind(&TaskDispatcherImpl::DispatcherThread, this))
{
    state_manager_sptr_->SetState(DispatcherStateManager::Starting);

    while (DispatcherStateManager::Starting == state_manager_sptr_->State())
        std::this_thread::sleep_for(std::chrono::milliseconds(10));
}

TaskDispatcherImpl::~TaskDispatcherImpl()
{
    state_manager_sptr_->SetState(DispatcherStateManager::Quiting);
    task_add_condition_.notify_all();
    while (DispatcherStateManager::Quitted != state_manager_sptr_->State())
        ;
    dispatcher_thread_.join();
}

bool TaskDispatcherImpl::AddTaskImpl(TaskSptr task, Task::TaskType type)
{
    if (likely(DispatcherStateManager::Started == state_manager_sptr_->State()))
    {
        task->SetState(Task::Waiting);
        task->SetTaskType(type);

        UniqueLocker locker(new_task_lock_);
        tasks_.emplace_back(task);
        task_add_condition_.notify_one();
        return true;
    }
    return false;
}

void TaskDispatcherImpl::DispatcherThread()
{
    state_manager_sptr_->SetState(DispatcherStateManager::Started);
    for (; DispatcherStateManager::Started == state_manager_sptr_->State();)
    {
        UniqueLocker locker(new_task_lock_);
        task_add_condition_.wait(locker);

        if (DispatcherStateManager::Quiting == state_manager_sptr_->State() /* waitting running tasks to finish. */)
        {
            break;
        }

        // Add a task to thread pool.
        if (unlikely(tasks_.empty()))
            continue;

        for (; !tasks_.empty();)
        {
            auto item = tasks_.front();
            tasks_.pop_front();
            if (!thread_pool::ThreadPool::GetInstance()->Start(item))
            {
                std::cerr << "\e[1;31m Start task failed. \e[0m";
                tasks_.emplace_back(item);
                continue;
            }
            item->SetState(Task::Running);
        }
    }
    state_manager_sptr_->SetState(DispatcherStateManager::Quitted);
}

}  // namespace dispatcher
